/*
 * Copyright (c) 2025 Li Auto Inc. and its affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef SRC_CPP_RTPS_TRANSPORT_UDSSTREAMTRANSPORTINTERFACE_H_
#define SRC_CPP_RTPS_TRANSPORT_UDSSTREAMTRANSPORTINTERFACE_H_

#include <transport/TransportInterface.h>
#include <transport/UDSStreamTransportDescriptor.h>
#include <deps/common/IPFinder.h>
#include <tcp/RTCPHeader.h>
#include "UDSStreamChannelResource.h"
#include "UDSStreamAcceptor.h"

#include <thread>
#include <vector>
#include <map>
#include <memory>
#include <mutex>
#include <string>

#include <asio.hpp>
#include <asio/steady_timer.hpp>

namespace vbs {
namespace transport {

class UDSStreamChannelResource;

class UDSStreamTransportInterface : public TransportInterface {

    class ReceiverInUseCV {
     public:
        bool in_use = false;

        std::condition_variable cv;
    };

    std::atomic<bool> alive_;

 public:
    RTPS_DllAPI UDSStreamTransportInterface(const UDSStreamTransportDescriptor&);

    virtual ~UDSStreamTransportInterface();

    bool CloseInputChannel(const Locator&) override;

    void CloseOutputChannel(std::shared_ptr<UDSStreamChannelResource>& channel);

    bool init() override;

    bool IsInputChannelOpen(const Locator&) const override;

    bool IsLocatorSupported(const Locator&) const override;

    bool OpenInputChannel(const Locator&, std::shared_ptr<ReceiverResource>, uint32_t) override;

    bool OpenOutputChannel(std::map<Locator_t, std::shared_ptr<SenderResource>>&, const Locator_t&,
                           Locator_t&) override;

    bool DoInputLocatorsMatch(const Locator&, const Locator&) const override;

    const UDSStreamTransportDescriptor* configuration() const;

    UDSStreamTransportDescriptor* configuration();

    void update_network_interfaces() override;

    bool Receive(std::shared_ptr<UDSStreamChannelResource>& channel, ertps::rtps::octet* receive_buffer,
                 uint32_t receive_buffer_capacity, uint32_t& receive_buffer_size, Locator& remote_locator,
                 RTCPHeader& rtcp_header);

    bool send(ertps::rtps::octet* send_buffer, uint32_t send_buffer_size,
              std::shared_ptr<UDSStreamChannelResource>& channel,
              ertps::rtps::LocatorsIterator* destination_locators_begin,
              ertps::rtps::LocatorsIterator* destination_locators_end);

    bool async_send(octet* send_buffer, uint32_t send_buffer_size, std::shared_ptr<UDSStreamChannelResource>& channel,
                    ertps::rtps::LocatorsIterator* destination_locators_begin,
                    ertps::rtps::LocatorsIterator* destination_locators_end, AsyncSendHandler handler);

    void SocketAccepted(std::shared_ptr<asio::local::stream_protocol::socket> socket, const Locator& locator,
                        const asio::error_code& error);

    void SocketConnected(const std::weak_ptr<UDSStreamChannelResource>& channel_weak_ptr,
                         const asio::error_code& error);

    uint32_t max_recv_buffer_size() const override { return configuration()->maxMessageSize; }

    UDSStreamTransportDescriptor* get_configuration() override { return &configuration_; }

    bool select_multicast() const override { return false; }

 private:
    asio::io_service io_service_;

    std::shared_ptr<std::thread> io_service_thread_;

    mutable std::mutex sockets_map_mutex_;
    mutable std::mutex receiver_resources_mutex_;
    mutable std::mutex send_resources_mutex_;
    mutable std::mutex acceptors_mutex_;

    std::map<Locator, std::shared_ptr<UDSStreamChannelResource>> channel_resources_;

    std::map<std::string, std::shared_ptr<ReceiverResource>> receiver_resources_;

    std::map<Locator, std::shared_ptr<UDSStreamAcceptor>> acceptors_;

    UDSStreamTransportDescriptor configuration_;

    void set_receive_buffer_size(uint32_t size);

    void set_send_buffer_size(uint32_t size);

    void endpoint_to_locator(const asio::local::stream_protocol::endpoint& endpoint, Locator& locator) const;

    bool is_locator_allowed(const Locator&) const override;

    bool send(ertps::rtps::octet* send_buffer, uint32_t send_buffer_size,
              std::shared_ptr<UDSStreamChannelResource>& channel, const Locator& remote_locator);

    bool async_send(ertps::rtps::octet* send_buffer, uint32_t send_buffer_size,
                    std::shared_ptr<UDSStreamChannelResource>& channel, const Locator& remote_locator,
                    AsyncSendHandler handler);

    void clean();

    bool is_input_path_open(std::string path) const;

    bool create_acceptor_socket(const Locator& locator);

    void close_uds_socket(std::shared_ptr<UDSStreamChannelResource>& channel);

    void perform_listen_operation(std::weak_ptr<UDSStreamChannelResource> channel_weak);

    bool read_body(ertps::rtps::octet* receive_buffer, uint32_t receive_buffer_capacity, uint32_t* bytes_received,
                   std::shared_ptr<UDSStreamChannelResource>& channel, std::size_t body_size);

    UDSStreamTransportInterface(int32_t transport_kind);
};

}  // namespace transport
}  // namespace vbs

#endif
